import asyncio
import json
import websockets.legacy.client
import sounddevice as sd
import numpy as np
import opuslib  # 恢复导入
from rich.console import Console
from rich.logging import RichHandler
import logging
from audio_player import AudioPlayer
import time
import signal
import sys
import select
from collections import defaultdict
from typing import Any, Callable, Dict, List, Optional
import concurrent.futures
import os
from iot_handler import IoTHandler  # 导入IoTHandler类
from audio_processor import AudioProcessor  # 导入AudioProcessor类
from message_handler import MessageHandler  # 导入MessageHandler类
import base64
import traceback
import datetime

# 设置日志
console = Console()
logging.basicConfig(
    level=logging.INFO,  # 改为INFO级别
    format="%(message)s",
    datefmt="[%X]",
    handlers=[RichHandler(console=console, rich_tracebacks=True)]
)
logger = logging.getLogger("xiaozhi")

# 设置websockets的日志级别为WARNING
logging.getLogger("websockets").setLevel(logging.WARNING)
# 设置asyncio的日志级别为WARNING
logging.getLogger("asyncio").setLevel(logging.WARNING)

def get_audio_device():
    """获取可用的音频输入设备"""
    try:
        # 获取所有设备
        devices = sd.query_devices()
        input_devices = []

        logger.info("\n可用的音频设备:")
        for i, device in enumerate(devices):
            if device['max_input_channels'] > 0:  # 只显示输入设备
                logger.info(f"{i}: {device['name']} (输入通道: {device['max_input_channels']})")
                input_devices.append((i, device))

        if not input_devices:
            logger.error("未找到任何音频输入设备！请确保已连接麦克风。")
            return None

        # 首先尝试找到最适合的设备（支持16000Hz采样率的麦克风）
        best_device = None
        for device_id, device in input_devices:
            if ('麦克风阵列' in device['name'] and
                    device['default_samplerate'] == 16000.0):
                best_device = device_id
                logger.info(f"找到最佳设备(16000Hz): {device['name']}")
                break

        # 如果没有找到最佳设备，尝试使用系统默认输入设备
        if best_device is None:
            try:
                default_device = sd.default.device[0]
                if default_device >= 0:  # 确保默认设备ID有效
                    device_info = sd.query_devices(default_device)
                    if device_info['max_input_channels'] > 0:
                        best_device = default_device
                        logger.info(f"使用系统默认输入设备: {device_info['name']}")
            except Exception as e:
                logger.debug(f"获取默认设备失败: {e}")

        # 如果还是没有设备，选择第一个麦克风设备
        if best_device is None:
            for device_id, device in input_devices:
                if '麦克风' in device['name']:
                    best_device = device_id
                    logger.info(f"使用第一个可用麦克风: {device['name']}")
                    break

        # 如果还是没有找到，使用第一个输入设备
        if best_device is None and input_devices:
            best_device = input_devices[0][0]
            device_info = sd.query_devices(best_device)
            logger.info(f"使用第一个可用输入设备: {device_info['name']}")

        if best_device is not None:
            return best_device
        else:
            logger.error("未找到合适的音频输入设备！")
            return None

    except Exception as e:
        logger.error(f"获取音频设备失败: {e}")
        logger.info("请检查以下几点：")
        logger.info("1. 确保系统已连接麦克风设备")
        logger.info("2. 检查系统声音设置中的输入设备是否正确配置")
        logger.info("3. 确保应用程序有权限访问麦克风")
        logger.info("4. 尝试在系统设置中禁用再启用麦克风")
        return None

def list_audio_devices():
    """列出所有音频设备的详细信息"""
    try:
        logger.info("\n=== 音频设备详细信息 ===")
        devices = sd.query_devices()

        # 获取默认设备
        try:
            default_input = sd.default.device[0]
            default_output = sd.default.device[1]
            logger.info(f"系统默认输入设备: {default_input}")
            logger.info(f"系统默认输出设备: {default_output}")
        except Exception as e:
            logger.debug(f"获取默认设备信息失败: {e}")

        # 显示所有设备的详细信息
        logger.info("\n所有设备列表:")
        for i, device in enumerate(devices):
            device_type = []
            if device['max_input_channels'] > 0:
                device_type.append("输入")
            if device['max_output_channels'] > 0:
                device_type.append("输出")
            type_str = " & ".join(device_type)

            logger.info(f"设备 {i}: {device['name']}")
            logger.info(f"    类型: {type_str}")
            logger.info(f"    输入通道: {device['max_input_channels']}")
            logger.info(f"    输出通道: {device['max_output_channels']}")
            logger.info(f"    采样率: {device['default_samplerate']}")

    except Exception as e:
        logger.error(f"列出音频设备失败: {e}")

def generate_test_audio():
    """生成测试音频（1秒钟的440Hz正弦波）"""
    sample_rate = 16000
    duration = 1.0  # 秒
    t = np.linspace(0, duration, int(sample_rate * duration))
    frequency = 440  # Hz
    amplitude = 0.5
    samples = (amplitude * np.sin(2 * np.pi * frequency * t)).astype(np.float32)
    return (samples * 32767).astype(np.int16)

# 客户端配置管理
class ClientConfig:
    def __init__(self, **kwargs):
        self.server_url = kwargs.get('server_url', 'ws://localhost:8000')
        self.device_id = kwargs.get('device_id', None)
        self.token = kwargs.get('token', None)
        self.client_id = kwargs.get('client_id', "xiaozhi-client-python")
        self.enable_audio_test = kwargs.get('enable_audio_test', True)
        self.enable_tools = kwargs.get('enable_tools', False)
        self.input_mode = kwargs.get('input_mode', 'audio')
        self.recording_mode = kwargs.get('recording_mode', 'auto')
        self.enable_audio_resend = kwargs.get('enable_audio_resend', False)
        self.enable_iot_register = kwargs.get('enable_iot_register', True)
        self.enable_iot_report = kwargs.get('enable_iot_report', True)

        # 音频参数
        self.audio = {
            'sample_rate': 16000,
            'channels': 1,
            'device_id': None,
            'frame_duration': 10  # 添加回 frame_duration 参数，默认为10ms
        }

        self.silence = {
            'threshold': kwargs.get('silence_threshold', 0.0001),
            'duration': kwargs.get('silence_duration', 1.5),
        }

        self.connection = {
            'server_url': kwargs.get('server_url'),
            'device_id': kwargs.get('device_id'),
            'token': kwargs.get('token'),
        }

        self.debug = {
            'enable_audio_test': self.enable_audio_test,  # 保持兼容性
        }

class XiaozhiClient:
    def __init__(self, server_url: str, device_id: str = None, token: str = None, client_id: str = None, enable_audio_test: bool = True, input_mode: str = "audio", recording_mode: str = "auto", one_shot_mode: bool = False):
        # 初始化配置
        self.config = ClientConfig(
            server_url=server_url,
            device_id=device_id,
            token=token,
            client_id=client_id,
            enable_audio_test=enable_audio_test,
            input_mode=input_mode,
            recording_mode=recording_mode
        )

        # 初始化日志
        self.logger = logging.getLogger("xiaozhi")

        # 初始化状态
        self.state = "idle"  # 初始状态为idle
        self.should_exit = False
        self.is_playing_audio = False
        self.tts_stop_received = False
        self.tts_started = False  # TTS开始状态标志
        self.tts_sentence_end_received = False  # 句子结束标志，用于触发重发

        # 静音检测状态记录
        self.last_silence_detected = False
        self.last_silence_detected_time = "未初始化"

        # 时间记录
        self.llm_response_time = 0

        # TTS缓存
        self.tts_cache = {}
        self.tts_cache_size = 20  # 最多缓存20条TTS
        self.current_tts_cache_key = None

        # 添加音频帧计数器
        self.audio_frame_counter = 0

        # 初始化音频设备（仅在音频模式下）
        if self.config.input_mode == "audio":
            # 显示详细的音频设备信息
            list_audio_devices()

            # 获取音频设备
            self.audio_device = get_audio_device()
            if self.audio_device is None:
                self.logger.error("无法初始化音频设备，请检查您的麦克风设置。")
                raise RuntimeError("无法初始化音频设备")

            # 验证设备是否支持指定的采样率
            try:
                device_info = sd.query_devices(self.audio_device, 'input')
                self.device_sample_rate = int(device_info['default_samplerate'])

                if self.device_sample_rate != self.config.audio['sample_rate']:
                    self.logger.warning(
                        f"设备采样率({self.device_sample_rate}Hz)与目标采样率({self.config.audio['sample_rate']}Hz)不匹配，将进行重采样"
                    )

                self.logger.info(f"已选择设备 {device_info['name']}")
                self.logger.info(f"设备采样率: {self.device_sample_rate}Hz")
                self.logger.info(f"目标采样率: {self.config.audio['sample_rate']}Hz")

                # 更新配置中的设备ID
                self.config.audio['device_id'] = self.audio_device

            except Exception as e:
                self.logger.error(f"验证设备参数失败: {e}")
                raise RuntimeError("设备参数验证失败")

            # 初始化音频处理器
            self.audio_processor = AudioProcessor(self.config)

            # 音频播放器
            self.audio_player = AudioPlayer(
                sample_rate=self.config.audio['sample_rate'],
                channels=self.config.audio['channels'],
                on_play_start=self.on_audio_play_start,
                on_play_end=self.on_audio_play_end,
                auto_start=False  # 不自动启动，等待IoT注册完成后再启动
            )

            # 音频数据队列
            self.audio_queue = asyncio.Queue()
        else:
            self.logger.info(f"使用文本输入模式，跳过音频设备初始化")
            # 在文本模式下，仍然需要音频播放器来播放TTS响应
            self.audio_player = AudioPlayer(
                sample_rate=self.config.audio['sample_rate'],
                channels=self.config.audio['channels'],
                on_play_start=self.on_audio_play_start,
                on_play_end=self.on_audio_play_end,
                auto_start=False  # 不自动启动，等待IoT注册完成后再启动
            )

        # 状态管理
        self.websocket = None
        self.loop = None

        # 添加响应事件
        self.response_received = asyncio.Event()

        # 添加退出标志
        self.should_exit = False

        # 添加退出事件
        self.exit_event = asyncio.Event()

        # 添加TTS停止标志
        self.tts_stop_received = False

        # 添加下一轮对话已开始标志
        self._next_round_started = False

        # 初始化IoT处理器
        self.iot_handler = IoTHandler(self.send_json)
        # 设置client引用
        self.iot_handler.set_client(self)

        # 初始化消息处理器
        self.message_handler = MessageHandler(
            logger=self.logger,
            send_json_callback=self.send_json,
            set_state_callback=self.set_state,
            audio_player=self.audio_player,
            iot_handler=self,
            client=self  # 直接在构造函数中传递客户端引用
        )

        # 一次性对话模式
        self.one_shot_mode = one_shot_mode
        if self.one_shot_mode:
            self.logger.info("已启用一次性对话模式：完成一轮对话后将自动结束")

        # 添加音频备份目录
        self.audio_backup_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "audio_backup")
        os.makedirs(self.audio_backup_dir, exist_ok=True)
        self.logger.info(f"音频备份目录: {self.audio_backup_dir}")

        # 音频重发控制
        self.resend_tasks = []
        self.resend_queue = asyncio.Queue()  # 新增: 用于存储所有音频数据以便重发

    async def handle_state_change(self, old_state: str, new_state: str):
        """处理状态变更"""
        self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 状态变更: {old_state} -> {new_state}")
        if self.websocket and self.websocket.open:
            message = {
                "type": "listen",
                "state": "start" if new_state == "recording" else "stop"
            }

            # 如果是开始录音状态，添加 mode 字段
            if new_state == "recording":
                message["mode"] = self.config.recording_mode

            await self.send_json(message)

    async def handle_audio_processed(self, audio_frames, volume: float):
        """处理处理后的音频数据，此方法主要用于在process_audio_queue中调用"""
        # 不需要再向队列添加内容，因为audio_callback已经处理了这个步骤
        # 这个方法现在主要用于通知其他可能依赖于音频处理的组件

        # 记录调试信息或触发事件
        self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 音频数据已处理，音量: {volume:.6f}")

        # 在这里可以添加其他需要的处理逻辑，如音量可视化等

    async def check_event_loop_blocking(self):
        """检查事件循环是否被阻塞"""
        start_time = time.time()
        # 创建一个简单的异步任务，测量它被调度执行的延迟
        await asyncio.sleep(0)
        end_time = time.time()
        delay = (end_time - start_time) * 1000
        self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] [时间跟踪] 事件循环延迟: {delay:.2f}ms")
        return delay

    async def handle_message(self, message: dict):
        """处理接收到的消息"""
        # 使用消息处理器处理消息
        await self.message_handler.handle_message(
            message=message,
            client_state=self.state,
            direct_stop_recording_callback=self.direct_stop_recording
        )

    async def handle_error(self, error: Exception, context: str = None):
        """处理错误"""
        self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 错误 ({context}): {str(error)}")
        if isinstance(error, websockets.exceptions.ConnectionClosed):
            await self._handle_connection_closed()

    def audio_callback(self, indata, frames, time_info, status):
        """音频输入回调"""
        if status:
            self.logger.warning(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 音频输入状态: {status}")

        try:
            # 只在recording状态下处理音频，并且不在播放音频时录音
            if self.state != "recording" or not self.websocket or not self.websocket.open or not self.loop or self.is_playing_audio:
                return

            # 处理音频数据
            try:
                current_time = time.time()
                opus_data, volume = self.audio_processor.process_audio(indata, self.device_sample_rate)
                # 确保audio_data是bytes类型
                if not isinstance(opus_data, bytes):
                    self.logger.warning(f"音频数据不是bytes类型: {type(opus_data)}")
                    return

                # 将音频数据和音量信息加入队列
                # 虽然删除了静音检测逻辑，但为保持接口一致，仍然传递is_silence参数（始终为False）
                data_package = (opus_data, volume, False, current_time)

                # 放入音频处理队列
                asyncio.run_coroutine_threadsafe(
                    self.audio_queue.put(data_package),
                    self.loop
                )

                # 同时将完整的数据包放入重发队列，完全模拟正常流程
                asyncio.run_coroutine_threadsafe(
                    self.resend_queue.put(data_package),
                    self.loop
                )

            except Exception as e:
                self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 音频处理错误: {e}")
                return

            # 记录音量日志
            if not hasattr(self, 'last_volume_log') or current_time - self.last_volume_log >= 1.0:
                self.last_volume_log = current_time
                self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 当前音量: {volume:.6f}")

        except Exception as e:
            self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 音频回调处理错误: {e}")
            self.logger.error("详细错误信息:", exc_info=True)

    async def set_state(self, new_state: str):
        """更新状态并通知服务器"""
        if new_state not in ["recording", "idle"]:
            self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 无效的状态: {new_state}")
            return

        if new_state == self.state:
            return  # 状态没有变化，不需要处理

        old_state = self.state
        self.state = new_state

        # 处理状态变更
        await self.handle_state_change(old_state, new_state)

    def get_headers(self) -> dict:
        """获取连接头信息"""
        headers = {
            # 按照服务端期望的格式设置请求头
            "protocol-version": "1",  # 修改为小写并使用服务端期望的版本号
            "device-id": self.config.connection['device_id'] or "unknown",  # 修改为小写
            "authorization": f"Bearer {self.config.connection['token']}" if self.config.connection['token'] else "",  # 修改为小写
            "input-mode": self.config.input_mode  # 修改为小写
        }
        # 如果有客户端ID，也添加到请求头中
        if hasattr(self.config, 'client_id') and self.config.client_id:
            headers["client-id"] = self.config.client_id

        return headers

    async def send_json(self, data: dict):
        """发送JSON数据"""
        if self.websocket and self.websocket.open and not self.should_exit:
            try:
                # 特别记录"listen"类型的消息，这表示录音状态变化
                if data.get("type") == "listen":
                    state = data.get("state", "unknown")
                    mode = data.get("mode", "未指定")
                    current_time = time.strftime('%H:%M:%S.%f')[:-3]
                    self.logger.info(f"[{current_time}] 发送录音状态消息到服务端: {json.dumps(data, ensure_ascii=False)}")

                    # 记录额外信息，如当前是否检测到静音
                    if state == "stop" and hasattr(self, 'last_silence_detected'):
                        silence_time = getattr(self, 'last_silence_detected_time', 'unknown')
                        silence_msg = f"最近一次静音检测: {silence_time}" if getattr(self, 'last_silence_detected', False) else "未检测到静音"
                        self.logger.info(f"[{current_time}] 停止录音状态（{silence_msg}）")

                # 发送消息
                await self.websocket.send(json.dumps(data, ensure_ascii=False))
            except websockets.exceptions.ConnectionClosed:
                self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 发送JSON时检测到连接已断开")
                await self._handle_connection_closed()
            except Exception as e:
                self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 发送JSON数据失败: {e}")
                if self.websocket is None or not self.websocket.open:
                    await self._handle_connection_closed()

    async def send_audio(self, audio_data: bytes):
        """发送音频数据"""
        if not audio_data or not self.websocket:
            return

        try:
            await self.websocket.send(audio_data)
        except websockets.exceptions.ConnectionClosed:
            self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 发送音频时检测到连接已断开")
            await self._handle_connection_closed()
        except Exception as e:
            self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 发送音频数据失败: {e}")
            if self.websocket is None or not self.websocket.open:
                await self._handle_connection_closed()

    async def process_audio_queue(self):
        """处理音频队列"""
        while not self.should_exit:
            try:
                # 等待队列中的音频数据
                opus_data, volume, is_silence, timestamp = await self.audio_queue.get()

                # 处理音频数据
                await self.handle_audio_processed(opus_data, volume)

                # 只在录音状态下发送音频数据
                if self.state == "recording":
                    # 发送音频数据到服务器
                    await self.send_audio(opus_data)

                    # 记录音频处理状态（仅用于日志）
                    self.last_silence_detected = False
                    self.last_silence_detected_time = time.strftime('%H:%M:%S.%f')[:-3]

            except asyncio.CancelledError:
                break
            except Exception as e:
                self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 处理音频队列出错: {e}")
                # 防止CPU占用过高
                await asyncio.sleep(0.1)

    async def start_recording(self):
        """开始录音"""
        if self.state == "recording" or self.is_playing_audio:
            return

        # 重置响应事件
        self.response_received.clear()

        # 重置TTS句子结束标志
        self.tts_sentence_end_received = False

        # 清空重发队列，准备新的录音
        self.resend_queue = asyncio.Queue()
        self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 重发队列已清空，准备新录音")

        try:
            # 启动音频输入流
            self.audio_stream = sd.InputStream(
                device=self.audio_device,  # 使用选定的设备
                samplerate=self.device_sample_rate,  # 使用设备原生采样率
                channels=self.config.audio['channels'],
                callback=self.audio_callback,
                dtype=np.int16,
                blocksize=int(self.device_sample_rate * self.config.audio['frame_duration'] / 1000)  # 根据设备采样率调整块大小
            )
            self.audio_stream.start()
            await self.set_state("recording")
            self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 开始录音 (使用设备: {sd.query_devices(self.audio_device)['name']})")
        except Exception as e:
            self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 启动录音失败: {e}")
            self.logger.error("详细错误信息:", exc_info=True)
            await self.set_state("idle")

    async def stop_recording(self):
        """停止录音"""
        start_time = time.time()
        if hasattr(self, 'stt_time'):
            self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 从识别结果到开始停止录音设备耗时: {(start_time - self.stt_time)*1000:.2f}ms")

        # 记录进入stop_recording方法的时间
        self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] [时间跟踪] 进入stop_recording方法")

        if self.state != "recording":
            self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 当前不在录音状态，无需停止")
            return

        # 记录状态检查完成的时间
        state_check_time = time.time()
        self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] [时间跟踪] 状态检查耗时: {(state_check_time - start_time)*1000:.2f}ms")

        # 1. 先停止录音设备
        if hasattr(self, 'audio_stream'):
            try:
                self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 开始停止录音设备...")
                device_stop_start = time.time()
                self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] [时间跟踪] 从方法开始到停止设备前耗时: {(device_stop_start - start_time)*1000:.2f}ms")

                # 停止录音设备
                self.audio_stream.stop()
                device_stop_time = time.time()
                self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] [时间跟踪] 停止录音设备操作耗时: {(device_stop_time - device_stop_start)*1000:.2f}ms")

                # 关闭录音设备
                self.audio_stream.close()
                device_close_time = time.time()
                self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] [时间跟踪] 关闭录音设备操作耗时: {(device_close_time - device_stop_time)*1000:.2f}ms")

                self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 停止录音设备耗时: {(time.time() - device_stop_start)*1000:.2f}ms")
            except Exception as e:
                self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 停止录音失败: {e}")

        # 2. 等待队列中的音频数据都发送完
        queue_start_time = time.time()
        self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] [时间跟踪] 从方法开始到处理队列前耗时: {(queue_start_time - start_time)*1000:.2f}ms")

        if not self.audio_queue.empty():
            queue_size = self.audio_queue.qsize()
            self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 等待发送剩余的 {queue_size} 帧音频数据...")

            # 记录队列处理开始时间
            queue_process_start = time.time()
            while not self.audio_queue.empty():
                await asyncio.sleep(0.05)  # 减少等待时间

                # 每200ms记录一次队列状态
                if time.time() - queue_process_start > 0.2:
                    remaining = self.audio_queue.qsize()
                    processed = queue_size - remaining
                    self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] [时间跟踪] 已处理 {processed}/{queue_size} 帧，剩余 {remaining} 帧")
                    queue_process_start = time.time()

            queue_end_time = time.time()
            self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 所有音频数据已发送完成，耗时: {(queue_end_time - queue_start_time)*1000:.2f}ms")
        else:
            self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] [时间跟踪] 音频队列为空，无需等待")
            queue_end_time = queue_start_time

        # 3. 确保最后一帧数据被服务器处理
        wait_start_time = time.time()
        self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] [时间跟踪] 从方法开始到等待最后一帧前耗时: {(wait_start_time - start_time)*1000:.2f}ms")

        await asyncio.sleep(0.2)  # 减少等待时间
        wait_end_time = time.time()
        self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 等待最后一帧处理耗时: {(wait_end_time - wait_start_time)*1000:.2f}ms")

        # 4. 最后才发送停止状态
        state_change_start = time.time()
        self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] [时间跟踪] 从方法开始到状态变更前耗时: {(state_change_start - start_time)*1000:.2f}ms")

        await self.set_state("idle")
        state_change_end = time.time()
        self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 状态变更耗时: {(state_change_end - state_change_start)*1000:.2f}ms")

        end_time = time.time()
        self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 停止录音完成，总耗时: {(end_time - start_time)*1000:.2f}ms")

        # 检查 stt_time 是否存在，避免报错
        if hasattr(self, 'stt_time'):
            self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] [时间跟踪] 从STT结果到停止录音完成总耗时: {(end_time - self.stt_time)*1000:.2f}ms")

        # 处理音频处理器缓冲区中的剩余数据
        if hasattr(self.audio_processor, 'flush_buffer'):
            remaining_frames = self.audio_processor.flush_buffer()
            for frame in remaining_frames:
                await self.audio_queue.put(frame)

        # 如果不是通过STT触发的停止，也执行重发逻辑
        # 判断是否需要执行重发 - 只在服务端未返回STT结果时执行
        if not hasattr(self, 'stt_time') or time.time() - getattr(self, 'stt_time', 0) > 10:
            self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 未检测到STT结果或STT结果已过期，执行重发逻辑")
            # 创建重发任务
            resend_task = asyncio.create_task(self._resend_queued_audio())
            # 将重发任务添加到任务列表中管理
            self.resend_tasks.append(resend_task)
            # 清理已完成的重发任务
            self._cleanup_completed_resend_tasks()

    async def abort(self):
        """中断当前对话"""
        await self.send_json({"type": "abort"})
        self.audio_player.stop()
        await self.stop_recording()
        self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 已中断当前对话")

    async def connect(self):
        """连接到服务器"""
        try:
            headers = self.get_headers()
            self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 连接到服务器，请求头: {headers}")

            # 使用 legacy.client.connect 以确保兼容性
            self.websocket = await websockets.legacy.client.connect(
                self.config.connection['server_url'],
                extra_headers=headers,
                # 移除 subprotocols 参数，使用服务端默认协议
                ping_interval=20,
                ping_timeout=30,
                close_timeout=30,
                compression=None
            )
            self.loop = asyncio.get_running_loop()

            # 启动消息接收任务
            self.receive_task = asyncio.create_task(self.receive_messages())

            # 只在音频模式下启动音频处理任务
            if self.config.input_mode == "audio":
                self.audio_process_task = asyncio.create_task(self.process_audio_queue())

            # 预加载常用TTS响应
            self.preload_task = asyncio.create_task(self.preload_common_tts())

            self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] WebSocket连接已建立，等待服务端握手消息")

            # 不立即注册回调，等待服务端的握手消息后再注册

            return True
        except Exception as e:
            self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 连接失败: {e}")
            return False

    async def preload_common_tts(self):
        """预加载常用TTS响应"""
        # 由于我们不再主动发送TTS请求，这个方法可以简化或移除
        self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 跳过TTS预加载，使用服务端自动TTS处理")

    async def receive_messages(self):
        """接收并处理服务器消息"""
        # 标记是否已收到握手消息
        handshake_received = False

        try:
            while not self.should_exit:  # 添加对should_exit的检查
                try:
                    # 添加对websocket的检查
                    if self.websocket is None or not self.websocket.open:
                        self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] Websocket已关闭，退出接收消息循环")
                        await self._handle_connection_closed()  # 确保调用连接关闭处理
                        break

                    # 使用wait_for添加超时，防止永久阻塞
                    try:
                        message = await asyncio.wait_for(self.websocket.recv(), timeout=0.5)  # 减少超时时间
                    except asyncio.TimeoutError:
                        # 超时后检查是否应该退出
                        if self.should_exit:
                            break
                        # 再次检查websocket状态
                        if self.websocket is None or not self.websocket.open:
                            self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 超时后检测到Websocket已关闭")
                            await self._handle_connection_closed()
                            break
                        continue

                    recv_time = time.time()

                    if isinstance(message, str):
                        # 处理JSON消息
                        try:
                            self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 收到原始消息: {message}")
                            data = json.loads(message)
                            msg_type = data.get("type", "unknown")
                            self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 收到服务端{msg_type}消息: {json.dumps(data, ensure_ascii=False)}")

                            # 处理握手消息
                            if msg_type == "hello" and not handshake_received:
                                handshake_received = True
                                self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 收到服务端握手消息，开始注册IoT设备能力")
                                if getattr(self.config, "enable_iot_register", True):
                                    self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 开始注册IoT设备能力:")
                                    capabilities = self.iot_handler.get_capabilities()
                                    for capability_id, capability in capabilities.items():
                                        self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] - {capability['name']}:")
                                        for command_id, command in capability["commands"].items():
                                            self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}]   - {command['name']}")
                                    self.iot_handler.enable_iot_register = self.config.enable_iot_register
                                    self.iot_handler.enable_iot_report = self.config.enable_iot_report
                                    await self.iot_handler.register_capabilities()
                                    self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 已发送IoT设备能力注册请求")
                                else:
                                    self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 配置禁用IoT能力注册，跳过注册流程")
                                self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 握手完成，连接已就绪")
                                await self._handle_hello_message(data)
                                continue

                            # 根据消息类型进行处理
                            if msg_type == "iot":
                                # 处理IoT消息
                                self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 收到IoT消息，开始处理")
                                await self.iot_handler.handle_message(data)
                            elif msg_type in ["llm", "tts", "stt"]:
                                # 立即处理LLM、TTS和STT消息
                                await self._handle_message(data)
                            else:
                                # 其他消息创建任务异步处理
                                asyncio.create_task(self._handle_message(data))

                        except json.JSONDecodeError as e:
                            self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] JSON解析错误: {e}, 原始消息: {message}")
                    elif isinstance(message, bytes):
                        # 处理音频数据
                        self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 收到二进制音频数据，长度: {len(message)}字节")
                        try:
                            # 使用_handle_audio_message方法处理二进制音频数据
                            await self._handle_audio_message(message)
                        except Exception as e:
                            self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 音频处理错误: {e}")
                    else:
                        self.logger.warning(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 未知消息类型: {type(message)}")
                except websockets.exceptions.ConnectionClosed:
                    self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 连接已关闭")
                    await self._handle_connection_closed()  # 确保调用连接关闭处理
                    break  # 直接跳出循环而不是重新抛出异常
                except Exception as e:
                    self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 处理消息时出错: {e}")
                    # 添加短暂休眠，防止错误循环过快消耗CPU
                    await asyncio.sleep(0.1)
                    # 检查是否应该退出
                    if self.should_exit:
                        break
                    # 检查websocket状态
                    if self.websocket is None or not self.websocket.open:
                        self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 错误处理后检测到Websocket已关闭")
                        await self._handle_connection_closed()
                        break

            self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 退出接收消息循环")

        except websockets.exceptions.ConnectionClosed:
            self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 小助手说再见啦，下次再聊~")
            await self._handle_connection_closed()  # 所有退出相关的操作都在这里处理
        except Exception as e:
            self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 接收消息时出错: {e}")
            await self._handle_connection_closed()  # 所有退出相关的操作都在这里处理

    async def _handle_connection_closed(self):
        """处理连接断开的情况"""
        try:
            self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] WebSocket连接已断开，准备退出程序...")

            # 1. 停止录音（仅在音频模式下）
            if self.config.input_mode == "audio":
                if self.state == "recording":
                    await self.stop_recording()
                else:
                    # 如果不在recording状态但有音频流，也要确保关闭
                    if hasattr(self, 'audio_stream'):
                        try:
                            self.audio_stream.stop()
                            self.audio_stream.close()
                        except Exception as e:
                            self.logger.warning(f"关闭音频流时出错: {e}")

            # 2. 停止音频播放
            if hasattr(self, 'audio_player') and self.audio_player:
                self.audio_player.stop()

            # 3. 清空音频队列（仅在音频模式下）
            if self.config.input_mode == "audio" and hasattr(self, 'audio_queue'):
                while not self.audio_queue.empty():
                    try:
                        self.audio_queue.get_nowait()
                        self.audio_queue.task_done()
                    except asyncio.QueueEmpty:
                        break

            # 4. 重置状态
            self.state = "idle"
            self.websocket = None

            # 5. 取消所有正在运行的任务
            if hasattr(self, 'audio_process_task') and self.config.input_mode == "audio":
                self.audio_process_task.cancel()
                try:
                    await asyncio.shield(self.audio_process_task)
                except (asyncio.CancelledError, Exception):
                    pass

            if hasattr(self, 'receive_task'):
                self.receive_task.cancel()
                try:
                    await asyncio.shield(self.receive_task)
                except (asyncio.CancelledError, Exception):
                    pass

            if hasattr(self, 'preload_task'):
                self.preload_task.cancel()
                try:
                    await asyncio.shield(self.preload_task)
                except (asyncio.CancelledError, Exception):
                    pass

            # 6. 关闭音频播放器
            if self.audio_player:
                self.audio_player.close()

            # 7. 设置退出标志和事件（统一在这里处理退出逻辑）
            self.should_exit = True
            self.exit_event.set()  # 通知主循环退出

            self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 资源清理完成，程序将退出")

        except Exception as e:
            self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 处理连接关闭时出错: {e}")
            # 即使清理过程出错，也要确保设置退出标志
            self.should_exit = True
            self.exit_event.set()

    def test_audio_playback(self):
        """测试音频播放系统"""
        # 如果是文本模式且禁用了音频测试，则跳过
        if self.config.input_mode == "text" and not self.config.enable_audio_test:
            self.logger.info(f"文本模式下且禁用了音频测试，跳过测试")
            return True

        # 文本模式下但启用了音频测试
        if self.config.input_mode == "text" and self.config.enable_audio_test:
            self.logger.info(f"文本模式下但启用了音频测试，执行音频测试以验证系统")

        try:
            # 确保音频播放器已初始化
            if not hasattr(self, 'audio_player') or self.audio_player is None:
                self.audio_player = AudioPlayer(
                    sample_rate=self.config.audio['sample_rate'],
                    channels=self.config.audio['channels']
                )

            self.logger.info(f"正在进行音频播放测试...")
            # 生成1秒的440Hz正弦波
            sample_rate = self.config.audio['sample_rate']
            duration = 1.0  # 秒
            t = np.linspace(0, duration, int(sample_rate * duration))
            frequency = 440  # Hz
            amplitude = 0.5
            samples = (amplitude * np.sin(2 * np.pi * frequency * t)).astype(np.float32)

            # 将float32转换为int16
            test_audio = (samples * 32767).astype(np.int16)

            # 直接添加到缓冲区，指定为非Opus格式
            self.logger.info(f"添加测试音频到缓冲区: {len(test_audio)} 样本")
            audio_bytes = test_audio.tobytes()
            self.audio_player.add_audio_data(audio_bytes, is_opus=False)  # 明确指定不是Opus格式

            # 等待音频播放完成
            self.logger.info(f"等待音频播放完成...")
            time.sleep(2.0)  # 增加等待时间到2秒

            self.logger.info(f"音频播放测试完成。如果您听到了一声蜂鸣音，说明音频系统工作正常。")
            return True
        except Exception as e:
            self.logger.error(f"音频播放测试失败: {e}")
            import traceback
            self.logger.error(f"详细错误: {traceback.format_exc()}")
            return False

    async def close(self):
        """关闭客户端"""
        self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 开始关闭客户端...")
        
        # 停止IoT状态上报任务
        try:
            await self.iot_handler.stop_status_reporting()
            self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] IoT状态上报任务已停止")
        except Exception as e:
            self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 停止IoT状态上报任务失败: {e}")

        # 设置退出标志
        self.should_exit = True

        # 取消接收消息任务
        if hasattr(self, 'receive_task') and self.receive_task and not self.receive_task.done():
            self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 取消接收消息任务")
            self.receive_task.cancel()
            try:
                await asyncio.wait_for(asyncio.shield(self.receive_task), timeout=0.5)
            except (asyncio.TimeoutError, asyncio.CancelledError):
                pass
            except Exception as e:
                self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 取消接收消息任务时出错: {e}")

        # 调用连接关闭处理
        try:
            await asyncio.wait_for(self._handle_connection_closed(), timeout=1.0)
        except asyncio.TimeoutError:
            self.logger.warning(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 处理连接关闭超时")
        except Exception as e:
            self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 处理连接关闭时出错: {e}")

        # 关闭websocket连接
        if self.websocket:
            try:
                await asyncio.wait_for(self.websocket.close(), timeout=1.0)
            except asyncio.TimeoutError:
                self.logger.warning(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 关闭websocket连接超时")
            except Exception as e:
                self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 关闭websocket连接时出错: {e}")
            finally:
                self.websocket = None

        # 确保音频播放器已关闭
        if hasattr(self, 'audio_player') and self.audio_player:
            try:
                self.audio_player.stop()
                self.audio_player.close()
            except Exception as e:
                self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 关闭音频播放器时出错: {e}")

        # 确保设置退出事件
        if hasattr(self, 'exit_event'):
            self.exit_event.set()

        self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 客户端关闭完成")

        # 取消所有重发任务
        for task in self.resend_tasks:
            if not task.done():
                task.cancel()
                try:
                    await task
                except asyncio.CancelledError:
                    pass
                except Exception as e:
                    self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 取消重发任务时出错: {e}")

        # 清空重发队列
        self.resend_tasks = []
        self.resend_queue = asyncio.Queue()

        # 重置TTS相关标志
        self.tts_started = False
        self.tts_sentence_end_received = False
        self.tts_stop_received = False

    def on_audio_play_start(self):
        """音频开始播放的回调"""
        callback_start = time.time()

        # 设置标志，避免重复调用
        if self.is_playing_audio:
            self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 已经在播放音频状态，跳过停止录音")
            return

        # 更新音频播放状态
        self.is_playing_audio = True

        # 在非主线程中调用异步方法，需要使用run_coroutine_threadsafe
        if self.loop and self.state == "recording":
            start_time = time.time()
            # 使用直接停止录音方法，避免复杂的队列处理逻辑
            future = asyncio.run_coroutine_threadsafe(self.direct_stop_recording(), self.loop)
            try:
                # 设置超时，避免永久等待
                future.result(timeout=0.5)
            except concurrent.futures.TimeoutError:
                self.logger.warning(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 停止录音任务超时")
            except Exception as e:
                self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 停止录音任务出错: {e}")

            self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 音频播放回调中停止录音耗时: {(time.time() - start_time)*1000:.2f}ms")

    def on_audio_play_end(self):
        """音频播放结束回调"""
        try:
            # 记录播放结束事件
            self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 检查TTS停止状态: client={self.tts_stop_received}, handler={getattr(self.message_handler, 'tts_stop_received', None)}")

            # 确保TTS停止标志一致性
            tts_stop_in_client = getattr(self, 'tts_stop_received', False)
            tts_stop_in_handler = False
            if hasattr(self, 'message_handler') and self.message_handler:
                tts_stop_in_handler = getattr(self.message_handler, 'tts_stop_received', False)

            # 如果任一标志为True，则两个标志都应该为True
            tts_has_stopped = tts_stop_in_client or tts_stop_in_handler

            # 保持停止状态的一致性，确保停止标志不会丢失
            if tts_has_stopped:
                # 设置两个地方的标志，确保一致性
                self.tts_stop_received = True
                if hasattr(self, 'message_handler') and self.message_handler:
                    self.message_handler.tts_stop_received = True

                # 确保音频播放器的标志也正确设置
                if hasattr(self, 'audio_player') and self.audio_player:
                    self.audio_player.tts_stop_received = True

                # 确保后续代码知道TTS已经停止
                self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] TTS已停止，准备下一轮对话")
            else:
                # 如果TTS没有停止，记录警告但仍然启动新对话 (修改)
                self.logger.warning(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 警告：音频播放结束但TTS停止标志未设置，仍然尝试启动新对话")
                # 强制设置TTS停止标志，确保可以启动新对话
                self.tts_stop_received = True
                if hasattr(self, 'message_handler') and self.message_handler:
                    self.message_handler.tts_stop_received = True
                if hasattr(self, 'audio_player') and self.audio_player:
                    self.audio_player.tts_stop_received = True

            # 检查是否开启了一次性对话模式
            if hasattr(self, 'one_shot_mode') and self.one_shot_mode:
                self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 检测到一次性对话模式，对话结束后将关闭连接")
                # 设置退出标志
                self.should_exit = True
                # 在事件循环中调用close方法关闭客户端
                if hasattr(self, 'loop') and self.loop and self.loop.is_running():
                    asyncio.run_coroutine_threadsafe(self.close(), self.loop)
                    self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 已调度客户端关闭任务")
                return

            # 防止重复启动下一轮
            if hasattr(self, '_next_round_starting') and self._next_round_starting:
                self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 已经在启动下一轮对话，跳过重复调用")
                return

            self._next_round_starting = True

            # 确保音频播放标志被重置
            self.is_playing_audio = False

            # 优先关闭可能存在的录音流
            if hasattr(self, 'audio_stream') and self.audio_stream and self.audio_stream.active:
                try:
                    self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 清理可能存在的录音流")
                    self.audio_stream.stop()
                    self.audio_stream.close()
                    self.audio_stream = None
                except Exception as e:
                    self.logger.warning(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 清理录音流时出错: {e}")

            # 强制设置状态为idle，确保能够启动新对话
            if hasattr(self, 'loop') and self.loop and self.loop.is_running():
                asyncio.run_coroutine_threadsafe(self.set_state("idle"), self.loop)

            # 启动下一轮对话
            self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] TTS已停止，立即开始新一轮对话")

            try:
                # 检查是否在主事件循环内
                try:
                    asyncio.get_running_loop()
                    is_in_event_loop = True
                except RuntimeError:
                    is_in_event_loop = False

                if is_in_event_loop:
                    # 如果在事件循环内，直接创建任务
                    self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 在事件循环内，直接创建任务")
                    asyncio.create_task(self._start_next_round_after_playback())
                else:
                    # 在非事件循环线程中，使用自定义的线程安全方法
                    self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 在非事件循环线程中，使用线程安全方法")
                    if hasattr(self, 'loop') and self.loop and self.loop.is_running():
                        self._schedule_next_round_in_event_loop()
                    else:
                        self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 无法获取事件循环，无法启动下一轮对话")
                        self._next_round_starting = False
            except Exception as e:
                self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 启动下一轮对话出错: {e}")
                self._next_round_starting = False
        except Exception as e:
            self.logger.error(f"音频播放结束回调出错: {e}")
            self.logger.error("详细错误信息:", exc_info=True)
            self._next_round_starting = False

    def _schedule_next_round_in_event_loop(self):
        """在事件循环中安全地调度下一轮对话的启动
        这是一个普通方法，可以在任何线程中调用
        """
        if not hasattr(self, 'loop') or not self.loop or not self.loop.is_running():
            self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 事件循环未运行，无法调度任务")
            self._next_round_starting = False
            return

        # 创建并提交任务到事件循环
        try:
            # 创建一个新的循环内执行函数
            async def event_loop_task():
                try:
                    self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 在事件循环内执行下一轮对话启动")
                    await self._start_next_round_after_playback()
                except Exception as e:
                    self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 执行下一轮对话时出错: {e}")
                    self.logger.error("详细错误:", exc_info=True)
                finally:
                    # 确保重置标志
                    self._next_round_starting = False

            # 使用call_soon_threadsafe提交一个会创建任务的回调函数
            self.loop.call_soon_threadsafe(
                lambda: asyncio.create_task(event_loop_task())
            )
            self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 已安全调度下一轮对话任务")
        except Exception as e:
            self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 调度任务时出错: {e}")
            self._next_round_starting = False

    async def _start_next_round_after_playback(self):
        """在音频播放结束后启动下一轮对话"""
        try:
            # 短暂等待确保状态更新
            await asyncio.sleep(0.05)  # 减少等待时间，提高响应性

            # 再次确认TTS是否真的停止了
            tts_stop_confirmed = getattr(self, 'tts_stop_received', False)

            # 记录当前状态
            self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 开始新对话前状态: TTS停止={tts_stop_confirmed}")

            if not tts_stop_confirmed:
                self.logger.warning(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] TTS未停止，取消开始新对话")
                # 重置标志
                if hasattr(self, '_next_round_starting'):
                    self._next_round_starting = False
                return "TTS未停止，取消开始新对话", self.state

            # 确认音频不再播放
            audio_still_playing = False
            if hasattr(self, 'audio_player') and self.audio_player:
                audio_still_playing = self.audio_player.is_audio_playing()

            if audio_still_playing:
                self.logger.warning(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 音频仍在播放，等待完成")
                # 等待音频自然结束，而不是强制停止
                await asyncio.sleep(0.5)  # 再等待一小段时间
                # 再次检查
                if self.audio_player and self.audio_player.is_audio_playing():
                    self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 音频播放持续时间过长，温和地停止播放")
                    # 使用非强制方式停止，允许缓冲区数据正常播放完成
                    self.audio_player.stop(force=False)

            # 启动下一轮对话
            await self._start_next_round()
            return "开始新一轮对话", self.state
        except Exception as e:
            self.logger.error(f"启动下一轮对话出错: {e}")
            self.logger.error("详细错误:", exc_info=True)
            return f"错误: {str(e)}", self.state
        finally:
            # 重置标志，无论成功或失败
            if hasattr(self, '_next_round_starting'):
                self._next_round_starting = False

    async def _start_next_round(self):
        """开始下一轮对话"""
        try:
            # 首先确保状态为idle，如果不是就设置为idle
            if self.state != "idle":
                self.logger.warning(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 开始新对话时状态不是idle而是 {self.state}，强制设置为idle")
                await self.set_state("idle")

            # 重置下一轮对话已开始标志
            self._next_round_started = False

            # 确保所有播放结束和TTS停止的标志都重置
            self.tts_stop_received = False
            if hasattr(self, 'message_handler') and self.message_handler:
                self.message_handler.tts_stop_received = False

            # 确保音频播放标志为False
            self.is_playing_audio = False

            # 根据输入模式决定行为
            if self.config.input_mode == "audio":
                # 音频模式：开始录音
                # 如果未启动录音或者录音启动失败，最多尝试3次
                for attempt in range(3):
                    # 检查当前是否有录音流正在运行
                    if hasattr(self, 'audio_stream') and self.audio_stream and self.audio_stream.active:
                        self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 检测到已有活动的录音流，先停止它")
                        try:
                            self.audio_stream.stop()
                            self.audio_stream.close()
                            await asyncio.sleep(0.1)  # 留出时间清理
                        except Exception as e:
                            self.logger.warning(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 停止已有录音流时出错: {e}")

                    self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 开始新一轮录音... (尝试 {attempt+1}/3)")
                    await self.start_recording()

                    # 验证录音是否成功启动
                    if hasattr(self, 'audio_stream') and self.audio_stream and self.audio_stream.active:
                        self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 录音成功启动")
                        break
                    else:
                        self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 录音启动失败，尝试重启")
                        await asyncio.sleep(0.2)  # 等待一小段时间再重试

                        # 最后一次尝试失败，记录严重错误
                        if attempt == 2:
                            self.logger.critical(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 三次尝试后仍无法启动录音，请检查音频设备是否可用")
            else:
                # 文本模式：等待用户输入
                self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 等待用户输入文本...")
                # 创建任务读取用户输入
                asyncio.create_task(self._read_user_input())
        except Exception as e:
            self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 启动新一轮对话时出错: {e}")
            self.logger.error("详细错误:", exc_info=True)
            # 尝试恢复到空闲状态
            await self.set_state("idle")

    async def direct_stop_recording(self):
        """直接停止录音，不创建新的任务"""
        if self.state != "recording":
            return

        start_time = time.time()

        # 在停止录音前，保存最后一帧音频数据以便可能的重发
        try:
            if hasattr(self, 'last_opus_data') and self.last_opus_data:
                # 记录这是由STT触发的停止，非静音检测
                self.last_silence_detected = False
                self.last_silence_detected_time = time.strftime('%H:%M:%S.%f')[:-3]

                self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 检测到STT结果，备份最后一帧音频数据")
                # 保存最后一帧音频数据并安排重发 - 这个方法不再需要，因为我们有了全量重发
                # await self._save_and_schedule_resend(self.last_opus_data)
        except Exception as e:
            self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 备份最后一帧音频数据失败: {e}")

        # 停止录音流
        if hasattr(self, 'audio_stream') and self.audio_stream and self.audio_stream.active:
            try:
                self.audio_stream.stop()
                self.audio_stream.close()
                self.audio_stream = None
            except Exception as e:
                self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 停止录音流失败: {e}")

        # 设置状态为idle - 这会发送{"type": "listen", "state": "stop"}到服务端
        await self.set_state("idle")

        # 在状态变更后，启动重发任务
        # 注意：必须在状态变更后执行，确保服务端先收到停止信号
        self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 状态已变更为idle，开始重发录音队列")
        resend_task = asyncio.create_task(self._resend_queued_audio())

        # 将重发任务添加到任务列表中管理
        self.resend_tasks.append(resend_task)

        # 清理已完成的重发任务
        self._cleanup_completed_resend_tasks()

    async def _handle_message(self, message):
        """处理收到的消息"""
        try:
            if isinstance(message, dict):
                # 直接处理已解析的字典消息
                message_type = message.get("type")
                # self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 开始处理消息类型: {message_type}")
                # self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 消息内容: {json.dumps(message, ensure_ascii=False)}")

                # 使用消息处理器处理消息
                await self.message_handler.handle_message(
                    message=message,
                    client_state=self.state,
                    direct_stop_recording_callback=self.direct_stop_recording
                )
            elif isinstance(message, str):
                # 解析JSON字符串消息
                try:
                    data = json.loads(message)
                    await self._handle_message(data)  # 递归处理解析后的消息
                except json.JSONDecodeError:
                    self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] JSON解析失败: {message}")
            elif isinstance(message, bytes):
                # 处理二进制音频数据
                self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 处理二进制音频数据，长度: {len(message)}字节")
                await self.message_handler.handle_audio_message(
                    audio_data=message,
                    client_state=self.state,
                    is_playing_audio=self.is_playing_audio,
                    direct_stop_recording_callback=self.direct_stop_recording
                )
            else:
                self.logger.warning(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 收到未知格式的消息: {type(message)}")

        except Exception as e:
            self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 处理消息时出错: {str(e)}")
            self.logger.error("详细错误信息:", exc_info=True)
            await self.handle_error(e, "处理消息时出错")

    async def _read_user_input(self):
        """读取用户输入的文本"""
        try:
            # 使用线程池执行器来处理阻塞的输入操作
            with concurrent.futures.ThreadPoolExecutor() as executor:
                # 提示用户输入
                user_input = await self.loop.run_in_executor(executor, input, "请输入您的消息: ")

                if user_input.strip():
                    # 发送文本消息
                    await self.send_text(user_input)
                else:
                    self.logger.info("输入为空，请重新输入")
                    # 重新启动输入任务
                    asyncio.create_task(self._read_user_input())
        except Exception as e:
            self.logger.error(f"读取用户输入时出错: {e}")
            # 确保状态为 idle
            if self.state != "idle":
                await self.set_state("idle")

    async def send_text(self, text: str):
        """发送文本消息"""
        try:
            if not self.websocket or not self.websocket.open:
                self.logger.error("WebSocket连接已断开，无法发送文本")
                await self._handle_connection_closed()
                return

            # 设置状态为处理中（这会自动发送 listen start 消息）
            await self.set_state("recording")

            # 不需要再发送开始监听消息，因为 set_state 已经发送了

            # 发送检测到语音消息（包含文本）
            detect_message = {
                "type": "listen",
                "state": "detect",
                "text": text
            }
            json_str = json.dumps(detect_message, ensure_ascii=False)
            await self.websocket.send(json_str)

            # 记录发送的消息
            self.logger.info(f"已发送文本消息: {text}")

            # 设置状态为空闲（这会自动发送 listen stop 消息）
            await self.set_state("idle")
        except Exception as e:
            self.logger.error(f"发送文本消息时出错: {e}")
            await self.set_state("idle")

    async def _handle_audio_message(self, message: dict):
        """处理音频消息"""
        try:
            if isinstance(message, dict) and message.get("type") == "audio":
                # 处理JSON格式的音频消息
                if message.get("status") == "start":
                    self.logger.info("收到音频开始消息")
                    self.audio_player.play_audio_stream()
                elif message.get("status") == "data":
                    audio_data = base64.b64decode(message.get("data", ""))
                    self.audio_player.add_audio_data(audio_data)
                elif message.get("status") == "end":
                    self.logger.info("收到音频结束消息")
                    self.audio_player.stop_audio_stream()
            else:
                # 处理二进制音频数据
                await self.message_handler.handle_audio_message(
                    audio_data=message,
                    client_state=self.state,
                    is_playing_audio=self.is_playing_audio,
                    direct_stop_recording_callback=self.direct_stop_recording
                )
        except Exception as e:
            self.logger.error(f"处理音频消息出错: {e}")
            traceback.print_exc()

    async def _handle_hello_message(self, message):
        """处理服务端的hello消息"""
        if message.get("audio_params"):
            # 从服务端获取音频参数
            audio_params = message.get("audio_params")
            self.logger.info(f"从服务端获取音频参数: {audio_params}")

            # 只在音频模式下且音频处理器存在时更新设置
            if self.config.input_mode == "audio" and hasattr(self, 'audio_processor') and self.audio_processor is not None:
                # 更新音频处理器的参数
                frame_duration = audio_params.get("frame_duration", 10)  # 默认10ms
                sample_rate = audio_params.get("sample_rate", 16000)

                # 确保update_frame_settings方法存在
                if hasattr(self.audio_processor, 'update_frame_settings'):
                    # 更新音频处理器的帧长设置
                    self.audio_processor.update_frame_settings(
                        frame_duration_ms=frame_duration,
                        sample_rate=sample_rate
                    )
                else:
                    self.logger.warning("音频处理器缺少update_frame_settings方法，无法更新音频参数")
            else:
                self.logger.info("文本模式下不更新音频处理器设置")

        # 在握手完成后启动音频播放器
        if hasattr(self, 'audio_player') and self.audio_player:
            try:
                self.logger.info("握手完成，正在启动音频播放器...")
                self.audio_player.start()
                self.logger.info("音频播放器启动完成")
            except Exception as e:
                self.logger.error(f"启动音频播放器时出错: {e}")
                import traceback
                self.logger.error(f"详细错误: {traceback.format_exc()}")
        
        # 启动IoT状态上报任务
        if getattr(self.config, "enable_iot_report", True):
            try:
                await self.iot_handler.start_status_reporting()
                self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] IoT状态上报任务已启动")
            except Exception as e:
                self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 启动IoT状态上报任务失败: {e}")
        else:
            self.logger.info("配置禁用IoT状态上报，跳过上报任务启动")

    async def _save_and_schedule_resend(self, opus_data: bytes):
        """
        保存Opus编码的音频数据到本地，并安排3秒后重发

        Args:
            opus_data: Opus编码的音频数据
        """
        # 首先检查是否启用了音频重发功能
        if not hasattr(self.config, 'enable_audio_resend') or not self.config.enable_audio_resend:
            self.logger.debug(f"音频重发功能已禁用，跳过保存和重发")
            return

        try:
            # 生成唯一的文件名
            timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S_%f")
            file_path = os.path.join(self.audio_backup_dir, f"audio_backup_{timestamp}.opus")

            # 保存Opus数据到文件
            with open(file_path, "wb") as f:
                f.write(opus_data)

            self.logger.info(f"已保存音频备份: {file_path} (大小: {len(opus_data)} 字节)")

            # 创建3秒后重发的任务
            resend_task = asyncio.create_task(self._resend_audio_after_delay(file_path, opus_data, 3.0))
            self.resend_tasks.append(resend_task)

            # 记录当前任务状态
            self.logger.info(f"已创建音频重发任务，当前共有 {len(self.resend_tasks)} 个待处理的重发任务")

            # 清理已完成的重发任务
            old_count = len(self.resend_tasks)
            self._cleanup_completed_resend_tasks()
            new_count = len(self.resend_tasks)
            if old_count != new_count:
                self.logger.info(f"已清理 {old_count - new_count} 个已完成的重发任务，剩余 {new_count} 个")

        except Exception as e:
            self.logger.error(f"保存音频备份失败: {e}")
            import traceback
            self.logger.error(f"详细错误: {traceback.format_exc()}")

    def _cleanup_completed_resend_tasks(self):
        """清理已完成的重发任务"""
        old_tasks = self.resend_tasks.copy()
        self.resend_tasks = [task for task in self.resend_tasks if not task.done()]

        # 检查是否有错误的任务
        for task in old_tasks:
            if task.done() and not task.cancelled():
                try:
                    # 获取任务结果，检查是否有异常
                    task.result()
                except Exception as e:
                    self.logger.error(f"重发任务执行失败: {e}")
                    import traceback
                    self.logger.error(f"任务错误详情: {traceback.format_exc()}")

    async def _resend_audio_after_delay(self, file_path: str, opus_data: bytes, delay: float):
        """
        等待指定延迟后重发音频数据

        Args:
            file_path: 音频备份文件路径
            opus_data: 原始Opus编码的音频数据
            delay: 延迟时间（秒）
        """
        # 再次检查是否启用了音频重发功能（以防配置在延迟期间被更改）
        if not hasattr(self.config, 'enable_audio_resend') or not self.config.enable_audio_resend:
            self.logger.debug(f"音频重发功能已禁用，取消重发任务")
            return

        try:
            self.logger.info(f"计划在 {delay} 秒后重发音频: {os.path.basename(file_path)}")

            # 等待指定时间
            await asyncio.sleep(delay)

            # 检查连接是否有效
            if not self.websocket or not self.websocket.open:
                self.logger.warning("无法重发音频: WebSocket连接已关闭")
                return

            # 重发前检查当前状态，避免干扰当前对话流程
            if self.state in ["recording", "processing"]:
                self.logger.warning(f"当前状态为 {self.state}，暂不重发音频以避免干扰")
                return

            self.logger.info(f"正在重发音频: {os.path.basename(file_path)}")

            # 读取文件并重发
            try:
                with open(file_path, "rb") as f:
                    saved_opus_data = f.read()
                await self.send_audio(saved_opus_data)
                self.logger.info("音频重发成功")
            except FileNotFoundError:
                # 如果文件不存在，使用内存中的数据
                self.logger.warning("音频备份文件不存在，使用内存中的备份数据重发")
                await self.send_audio(opus_data)

            # 可选：发送完成后删除备份文件
            try:
                os.remove(file_path)
                self.logger.debug(f"已删除音频备份文件: {file_path}")
            except Exception as e:
                self.logger.warning(f"删除音频备份文件失败: {e}")

        except Exception as e:
            self.logger.error(f"重发音频失败: {e}")

    async def _resend_queued_audio(self):
        """
        循环发送重发队列中的所有数据，完全模拟正常流程
        在收到STT消息并停止录音后调用此方法
        """
        # 检查是否启用了音频重发功能
        if not hasattr(self.config, 'enable_audio_resend') or not self.config.enable_audio_resend:
            self.logger.debug(f"音频重发功能已禁用，跳过队列音频重发")
            return

        self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 开始重发录音队列中的数据")

        # 获取队列大小
        queue_size = self.resend_queue.qsize()
        if queue_size == 0:
            self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 重发队列为空，无需重发")
            return

        self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 重发队列中有 {queue_size} 帧数据等待重发")

        # 等待服务端发送第一条句子结束消息后再重发
        self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 等待服务端第一条句子结束消息...")

        # 重置句子结束标志，确保是新的触发
        self.tts_sentence_end_received = False

        # 智能等待，直到收到句子结束消息或超时
        tts_wait_start = time.time()
        while not self.tts_sentence_end_received and time.time() - tts_wait_start < 300.0:  # 最多等待3秒
            await asyncio.sleep(0.1)  # 每100ms检查一次状态

        # 判断等待结果
        if self.tts_sentence_end_received:
            self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 已收到第一条句子结束消息，立即开始重发音频")
        else:
            self.logger.warning(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 等待句子结束消息超时(3秒)，继续重发")

        # 创建一个临时列表来存储所有数据
        audio_frames = []
        while not self.resend_queue.empty():
            try:
                data_package = await self.resend_queue.get()
                audio_frames.append(data_package)
            except asyncio.CancelledError:
                break
            except Exception as e:
                self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 获取重发数据失败: {e}")

        # 显示将要重发的总帧数
        total_frames = len(audio_frames)
        self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 准备重发 {total_frames} 帧音频数据")

        # 按照原始顺序重发数据
        for i, data_package in enumerate(audio_frames):
            try:
                # 解包数据
                opus_data, volume, is_silence, timestamp = data_package

                # 完全模拟处理流程
                await self.handle_audio_processed(opus_data, volume)

                # 发送音频数据
                await self.send_audio(opus_data)

                # 每发送100帧记录一次进度
                if (i+1) % 100 == 0 or i+1 == total_frames:
                    self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 重发进度: {i+1}/{total_frames} 帧")

                # 控制发送速率，避免过快发送
                if i % 10 == 0:  # 每10帧暂停一次
                    await asyncio.sleep(0.01)  # 短暂暂停10ms

            except asyncio.CancelledError:
                self.logger.warning(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 重发过程被取消，已发送 {i+1}/{total_frames} 帧")
                break
            except Exception as e:
                self.logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 重发第 {i+1} 帧失败: {e}")
                # 继续尝试发送其他帧

        self.logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 音频数据重发完成")

        # 重发完成后，清除last_opus_data，避免下次STT触发时重发旧数据
        if hasattr(self, 'last_opus_data'):
            self.last_opus_data = None
            self.logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 已清除last_opus_data")

async def main():
    # 创建退出事件
    exit_event = asyncio.Event()

    def signal_handler(signum, frame):
        """信号处理函数"""
        logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 收到退出信号，准备关闭程序...")
        # 在主线程中设置事件
        asyncio.get_event_loop().call_soon_threadsafe(exit_event.set)

    # 注册信号处理器（对于支持的平台）
    try:
        if sys.platform != 'win32':
            signal.signal(signal.SIGINT, signal_handler)
            signal.signal(signal.SIGTERM, signal_handler)
    except Exception as e:
        logger.debug(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 注册信号处理器失败（这在Windows上是正常的）: {e}")

    # 加载配置文件
    config = {}
    try:
        # 支持通过命令行参数选择配置文件
        config_index = sys.argv[1] if len(sys.argv) > 1 else ""
        if config_index:
            config_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), f"config{config_index}.json")
        else:
            config_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.json")
        if os.path.exists(config_path):
            with open(config_path, "r", encoding="utf-8") as f:
                config = json.load(f)
                logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 已加载配置文件: {config_path}")
        else:
            logger.warning(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 配置文件不存在，使用默认配置: {config_path}")
    except Exception as e:
        logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 加载配置文件失败: {e}")

    # 从配置文件获取设置
    server_url = config.get("server_url", "ws://127.0.0.1:8000/xiaozhi/v1")
    device_id = config.get("device_id", "test-device")
    token = config.get("token", "test-token")
    client_id = config.get("client_id", "test-client")
    enable_audio_test = config.get("enable_audio_test", False)
    enable_tools = config.get("enable_tools", False)
    enable_audio_resend = config.get("enable_audio_resend", False)  # 新增: 获取音频重发功能配置
    input_mode = config.get("input_mode", "audio")  # 默认为音频模式

    # 创建客户端实例
    client = XiaozhiClient(
        server_url=server_url,
        device_id=device_id,
        token=token,
        client_id=client_id,
        enable_audio_test=enable_audio_test,
        input_mode=input_mode,
        recording_mode="auto",  # 使用自动模式
        one_shot_mode=False  # 不启用一次性对话模式
    )

    # 设置额外的配置参数
    client.config.enable_tools = enable_tools
    client.config.enable_audio_resend = enable_audio_resend  # 设置音频重发功能
    client.config.enable_iot_register = config.get("enable_iot_register", True)
    client.config.enable_iot_report = config.get("enable_iot_report", True)
    logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 音频重发功能: {'启用' if enable_audio_resend else '禁用'}")

    # 只在启用测试的情况下进行音频测试
    if client.config.enable_audio_test:
        if not client.test_audio_playback():
            logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 音频系统测试失败，请检查您的音频设备设置。")
            return
    else:
        logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 音频测试已禁用，跳过测试")

    try:
        # 连接服务器
        if not await client.connect():
            return

        # 根据输入模式决定初始行为
        if input_mode == "audio":
            # 直接开始录音
            await client.start_recording()
            logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 已开始自动录音模式，检测到1.5秒静音将自动停止...")
        else:
            # 等待用户输入文本
            logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 已启动文本输入模式，请输入您的消息...")
            asyncio.create_task(client._read_user_input())

        # 创建任务来等待事件
        client_exit_task = asyncio.create_task(client.exit_event.wait())
        system_exit_task = asyncio.create_task(exit_event.wait())

        # 等待任意一个事件发生
        done, pending = await asyncio.wait(
            [client_exit_task, system_exit_task],
            return_when=asyncio.FIRST_COMPLETED
        )

        # 取消未完成的任务
        for task in pending:
            task.cancel()

        logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 收到退出信号，准备关闭程序...")

    except asyncio.CancelledError:
        logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 程序被取消")
    except Exception as e:
        logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 程序出错: {str(e)}")
        import traceback
        logger.error(f"错误详情:\n{traceback.format_exc()}")
    finally:
        # 设置退出标志，确保所有循环都能正确退出
        client.should_exit = True

        # 快速关闭客户端
        try:
            await asyncio.wait_for(client.close(), timeout=2.0)  # 增加超时时间
        except (asyncio.TimeoutError, Exception) as e:
            logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 关闭客户端时出错: {str(e)}")

        # 确保程序退出
        logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 程序即将退出")
        # 强制退出，确保没有僵尸进程
        os._exit(0)

if __name__ == "__main__":
    # 根据平台设置合适的事件循环策略
    if sys.platform == 'win32':
        # Windows下使用selector事件循环
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    elif sys.platform == 'darwin':
        # macOS下使用默认策略，但确保使用selector
        policy = asyncio.get_event_loop_policy()
        policy._loop_factory = asyncio.SelectorEventLoop

    try:
        # 使用更健壮的方式运行主函数
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)

        try:
            loop.run_until_complete(main())
        except KeyboardInterrupt:
            logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 程序被用户中断")
        finally:
            # 确保事件循环正确关闭
            try:
                # 取消所有未完成的任务
                pending = asyncio.all_tasks(loop)
                for task in pending:
                    task.cancel()

                # 等待所有任务完成取消
                if pending:
                    loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True))
            except Exception as e:
                logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 清理任务时出错: {e}")
            finally:
                loop.close()
    except Exception as e:
        logger.error(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 程序异常退出: {e}")
        import traceback
        logger.error(f"错误详情:\n{traceback.format_exc()}")
    finally:
        logger.info(f"[{time.strftime('%H:%M:%S')}.{int(time.time()*1000)%1000:03d}] 程序已退出")
        # 强制退出，确保没有僵尸进程
        os._exit(0)